[fix](pipeline) Add streaming aggregation local exchange switch - #66216
Conversation
|
run buildall |
|
/review |
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
24bcb95 to
d1eeddf
Compare
|
run buildall |
|
/review |
d1eeddf to
8808186
Compare
|
run buildall |
|
/review |
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
FE UT Coverage ReportIncrement line coverage |
There was a problem hiding this comment.
Request changes: one P2 issue remains.
The new enable_local_exchange_before_streaming_agg session variable is not included in follower-to-master session forwarding. For forwarded SELECT or INSERT statements, an explicit true is lost and the master sends the default false to BE, so the switch silently cannot enable the requested exchange.
Critical checkpoints:
- Distribution and lifecycle: the default-false inherited distribution is safe for partial, non-finalizing
StreamingAggOperatorX; downstream shuffle propagation, task-local state, hash-join passthrough precedence, empty-partition fallback, pushed Top-N, and query-cache paths were checked with no additional defect. - FE/Thrift/BE contract: declaration, persistence and cloning, randomization, Thrift field ID 226, Java setter/
__isset, absent-field false default, and mixed-version unknown-field behavior were checked. FE statement forwarding is the blocking gap. - Coverage and validation: no builds or tests were run by this review runner, per the review-only instruction. Existing tests do not directly cover the switch matrix or the FE forwarding round trip; the inline comment requests targeted forwarding coverage. At payload freeze, style, formatter, and Cloud UT checks passed, while compile and FE/BE unit checks were still pending.
- Existing context and user focus: there were no existing inline review threads and no additional user-provided focus; the complete PR was reviewed.
| @VariableMgr.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_AGG, fuzzy = true) | ||
| public boolean enableLocalExchangeBeforeAgg = true; | ||
|
|
||
| @VariableMgr.VarAttr(name = ENABLE_LOCAL_EXCHANGE_BEFORE_STREAMING_AGG, fuzzy = true) |
There was a problem hiding this comment.
[P2] Forward the new switch to the master FE
Please add needForward = true to this VarAttr. FEOpExecutor forwards only SessionVariable.getForwardVariables(), whose filter excludes this field as currently annotated; ConnectProcessor then reconstructs the master-side session from that filtered map. Whenever a follower forwards a SELECT or INSERT (for example because it cannot read), enable_local_exchange_before_streaming_agg=true is therefore dropped and SessionVariable.toThrift() sends the master's default false, so the switch silently has no effect. Please cover this forwarding round trip as well.
There was a problem hiding this comment.
Follow-up review: no new nonduplicate inline findings.
The existing P2 at discussion_r3672330401 remains valid and blocking: the new
session switch needs needForward = true plus a follower-to-master forwarding
round-trip test. I did not repeat that inline comment.
Critical checkpoints:
- Distribution and lifecycle: every reachable
StreamingAggOperatorXis a
non-finalizing partial aggregate with grouping keys. Default false safely
inherits the current distribution; explicit true requests hash shuffle;
direct hash-join passthrough, serial-child inheritance, downstream
hash/bucket propagation, query-cache wrapping, and group-limit paths retain
their existing contracts. - FE/Thrift/BE contract: SET, persistence, cloning, fuzzy initialization,
toThrift, field ID 226, Java/BE names, false defaults, absent fields, and
mixed-version unknown-field handling are consistent. Follower forwarding is
the one gap already covered by the existing thread. - Compatibility and scope: the definition-only legacy
shuffled_agg_node_idspath and the partitioned hash-join wrapper behavior
both pre-exist at the authoritative base and are not introduced by this PR. - Coverage and validation: this review-only runner performed no builds or
tests. Direct switch-matrix and optional-field tests are absent; only the
missing forwarding round trip corresponds to a substantiated defect. - Existing context and user focus: all existing review context was
deduplicated, and there was no additional user-provided focus.
Review completion: complete for
880818677ff0db6066c0a4e5e4e828536a115424; no unresolved new candidate
remains, but the already-requested forwarding fix is still required.
There was a problem hiding this comment.
Final convergence verification for 8808186: request changes remains appropriate. The existing P2 forwarding defect is the only blocker; no new inline comment is added because it is already covered at discussion_r3672330401. A concurrent body-only follow-up review (4806100152) reached the same conclusion.
Critical checkpoints:
- Distribution and lifecycle: the default-false path safely inherits upstream distribution for the reachable non-finalizing StreamingAggOperatorX; explicit true requests the expected hash exchange, while hash-join passthrough precedence, empty grouping, downstream shuffle propagation, and serial/task safeguards remain sound.
- FE/Thrift/BE contract: declaration, reflective SET registration, persistence and cloning, toThrift, unique optional field 226, RuntimeState copy/getter behavior, absent-field false semantics, and mixed-version unknown-field handling are sound. Follower-to-master session forwarding remains the blocking gap.
- Coverage and validation: this review runner performed no builds or tests, as required. Existing tests do not directly cover the switch matrix or follower forwarding round trip; the existing inline thread requests targeted coverage.
- Existing context and user focus: the known forwarding thread was deduplicated, no distinct additional issue was found, and there was no additional user-provided focus. The complete four-file PR was reviewed.
- Review completion: the repository review skill was unavailable; the authoritative bundle prompt was followed as the fallback. Two normal full-coverage reviewers and the separate risk-focused reviewer converged with NO_NEW_VALUABLE_FINDINGS.
FE Regression Coverage ReportIncrement line coverage |
What
Add
enable_local_exchange_before_streaming_agg, defaulting tofalse, and use it to control whether streaming aggregation requests a local hash exchange.Why
The historical
QueryContext::should_be_shuffled_agg()interface is no longer available on branch-4.1. Reusingenable_local_exchange_before_aggalso changes streaming aggregation whenever the generic aggregation switch is enabled. A dedicated switch preserves the existing streaming aggregation distribution by default while allowing the local exchange to be enabled explicitly.Validation
git diff --check